package storm;
// @Author   :陈彤
import bean.order;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import redis.clients.jedis.Jedis;


import java.util.List;
import java.util.Map;

/**
 * @author chentong
 * @date 2021年12月10日 20：55
 */
public class LogFileBolt extends BaseRichBolt {
    //    private static final long serialVersionUID = 886149197481637894L;
    private OutputCollector collector;
    public static JRedisUtil jedis = new JRedisUtil("192.168.52.140",6379);//127.0.0.1
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    public void execute(Tuple tuple) {
        order order = new order();
        String orderInfo = (String) tuple.getValue(0);
        //System.out.println(orderInfo);
        order = spliteFile.getOrdersBean(orderInfo);
            System.out.println("---------1----------");
            System.out.println(order.getBookNum());
            System.out.println("--------------------");

        try {
            //订单数量
            jedis.incrBy("OrderNum",new Long(1));
            //销量排名前十的图书
            jedis.zincrby("BookTop",order.getBookNum(),order.getBookName());
            //销量排名前十的出版社
            jedis.zincrby("PressTop",order.getBookNum(),order.getBookPress());
        }catch (Exception e){
            System.out.println(orderInfo);
            e.printStackTrace();
        }finally {
//            System.out.println("---------1----------");
//            System.out.println(order.getBookName());
//            //System.out.println(jedis.get("orderNum"));
//            System.out.println("--------------------");
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //declarer.declare(new Fields(new String[]{"order"}));
    }
}